package spring.cloud.tasks.executor.executor;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.utils.CloseableExecutorService;
import spring.cloud.tasks.common.ExecutorServiceUtil;
import spring.cloud.tasks.common.ThreadFactory;
import spring.cloud.tasks.common.curator.Curator;
import spring.cloud.tasks.common.nodepaths.EnabledTaskNodePaths;
import spring.cloud.tasks.executor.executor.task.task_item_allot.TaskItemAssignListener;

import java.util.concurrent.Executors;

@Slf4j
@Setter
@Getter
public class TaskExecutorInitAndTaskItemAssignListenerService {

    public Curator curator;
    public String executorId;
    public Executor executor;
    //

    public TreeCache taskExecutorTreeCache;
    public java.util.concurrent.ExecutorService taskExecutorExecutorService;
    public NodeCache taskItemAssignListenerNodeCache;

    //
    public TaskExecutorInitAndTaskItemAssignListenerService(Executor executor, ExecutorInitAndTaskInitService executorInitAndTaskInitService) {
        this.curator = executorInitAndTaskInitService.getCurator();
        this.executorId = executorInitAndTaskInitService.getExecutorId();
        this.executor = executor;
    }

    public void start() throws Exception {
        String enableTasksRootPath = EnabledTaskNodePaths.ROOT;
        taskExecutorTreeCache = TreeCache.newBuilder(curator.getCuratorFramework(), enableTasksRootPath).//
                setCacheData(true).//
                setExecutor(new CloseableExecutorService(Executors.newSingleThreadExecutor(new ThreadFactory(executorId + "-task-enable-watcher", false)), true))//
                .setMaxDepth(1)//
                .build();
        taskExecutorExecutorService = Executors.newSingleThreadExecutor(new ThreadFactory(executorId + "-task-executor-init-thread", false));
        taskExecutorTreeCache.getListenable().addListener(new TaskExecutorInitListener(this), taskExecutorExecutorService);
        taskExecutorTreeCache.start();
        //
        taskItemAssignListenerNodeCache = new NodeCache(curator.getCuratorFramework(), "/task_item_allot");
        //添加节点监听事件,NodeCacheListener每次都会触发，但不能获取监听的操作类型到底是添加还是删除等。
        taskItemAssignListenerNodeCache.getListenable().addListener(new TaskItemAssignListener(taskItemAssignListenerNodeCache, executor));
        taskItemAssignListenerNodeCache.start(true);
    }

    public void shutdown() {
        try {
            if (taskItemAssignListenerNodeCache != null) {
                taskItemAssignListenerNodeCache.close();
            }
        } catch (Throwable t) {
            log.info(t.toString(), t);
        }
        try {
            if (taskExecutorTreeCache != null) {
                taskExecutorTreeCache.close();
            }
        } catch (Throwable t) {
            log.info(t.toString(), t);
        }
        try {
            ExecutorServiceUtil.shutdownNow(taskExecutorExecutorService);
        } catch (Throwable t) {
            log.info(t.toString(), t);
        }
    }


}
